package com.behavioranalysis.flinkprogram.flink.ad;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple5;

import java.text.SimpleDateFormat;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午9:18
 * updated by yang on 20-2-14 下午9:18
 */
public class TimeMinuteTuple2Tuple5 implements
        MapFunction<
                Tuple2<String, Long>,
                Tuple5<String, String, String, Long, Long>> {
    private static final long serialVersionUID = 1L;
    @Override
    public Tuple5<String, String, String, Long, Long> map(
            Tuple2<String, Long> value) throws Exception {
        SimpleDateFormat dateFormat =
                new SimpleDateFormat("yyyyMMdd");
        SimpleDateFormat dateFormat1 =
                new SimpleDateFormat("yyyy-MM-dd");

        String[] keySplited = value.f0.split("_");
        long adid = Long.valueOf(keySplited[1]);
        long clickCount = value.f1;
        String dateMinute = keySplited[0];

        String date = dateFormat1.format(dateFormat.parse(
                dateMinute.substring(0, 8)));
        String hour = dateMinute.substring(8, 10);
        String minute = dateMinute.substring(10);

        return new Tuple5<>(date, hour, minute, adid, clickCount);
    }
}
